package cn.tukk.sparkproject.spark.product.MyUDF;

import org.apache.spark.sql.Row;
import org.apache.spark.sql.expressions.MutableAggregationBuffer;
import org.apache.spark.sql.expressions.UserDefinedAggregateFunction;
import org.apache.spark.sql.types.DataType;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructType;

import java.util.Arrays;

public class MyGroupConcatDistinctUDAF  extends UserDefinedAggregateFunction {

    private static final long serialVersionUID = 5367469190866395373L;
    //指定输入数据的字段与类型
    // 指定输入数据的字段与类型
    private StructType inputSchema = DataTypes.createStructType(Arrays.asList(DataTypes.createStructField("cityInfo",DataTypes.StringType,true)));
    //指定缓冲数据的字段与类型
    private StructType bufferSchema=DataTypes.createStructType(Arrays.asList(DataTypes.createStructField("bufferCityInfo",DataTypes.StringType,true)));
    //指定返回类型
    private DataType dataType=DataTypes.StringType;
    //指定是否是确定性的
    private boolean deterministic=true;

    @Override
    public StructType inputSchema() {
        return inputSchema;
    }

    @Override
    public StructType bufferSchema() {
        return bufferSchema;
    }

    @Override
    public DataType dataType() {
        return dataType;
    }

    @Override
    public boolean deterministic() {
        return deterministic;
    }

    /**
     * 初始化，内部指定一个初始化的值
     * @param buffer
     */
    @Override
    public void initialize(MutableAggregationBuffer buffer) {
        buffer.update(0,"");
    }

    /**
     * 更新，可以认为是，一个一个将组内字符串传递过来  拼接
     * @param buffer
     * @param input
     */
    @Override
    public void update(MutableAggregationBuffer buffer, Row input) {
        //拼接中的已经拼接过的城市信息串
        String bufferCityInfo = buffer.getString(0);
        //刚刚传递进来的某个城市信息
        String cityInfo = input.getString(0);
        //在这里实现去重逻辑
        //判断之前没有拼接过某个城市信息，才可以继续拼接
        if(!bufferCityInfo.contains(cityInfo)){
            if("".equals(bufferCityInfo)){
                bufferCityInfo+=cityInfo;
            }else {
                //比如 1：北京
                //1：北京，2：上海
                bufferCityInfo+=","+cityInfo;
            }
            buffer.update(0,bufferCityInfo);
        }
    }

    /**
     *
     *    合并update操作，可能是针对一个分组内部分数据，在某个将诶点上发生的
     *    但是的可能一个分组内的数据，会分布在多个节点上处理 此时就要merge操作，将各个节点分布式拼接好的串，合并起来
     * @param buffer
     * @param input
     */
    @Override
    public void merge(MutableAggregationBuffer buffer, Row input) {
        String bufferCityInfo1 = buffer.getString(0);
        String bufferCityInfo2 = input.getString(0);

        for(String cityInfo : bufferCityInfo2.split(",")) {
            if(!bufferCityInfo1.contains(cityInfo)) {
                if("".equals(bufferCityInfo1)) {
                    bufferCityInfo1 += cityInfo;
                } else {
                    bufferCityInfo1 += "," + cityInfo;
                }
            }
        }

        buffer.update(0, bufferCityInfo1);



    }

    @Override
    public Object evaluate(Row buffer) {
        return buffer.getString(0);
    }
}
